package com.gblfy.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class MyProducer1 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Map<String, Object> configs = new HashMap<String, Object>();
        //指定初始化连接用到的broker地址
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.92.104:9092");
        //指定key序列化类
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        //指定value序列化类
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        //构造生产者对象 指定发送的key和value的类型 配置的参数列表（必填参数+辅助参数）
        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(configs);

        //用于设置用户自定义的消息头字段
        List<Header> headers = new ArrayList<>();
        headers.add(new RecordHeader("biz.name", "producer.demo".getBytes()));

        //构造record封装发送消息主体
        ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(
                "topic_1", //指定发送主题
                0,//指定发送分区
                5,//指定发送key
                "hello gblfy 0",//指定发送消息主题
                headers//用于设置用户自定义的消息头字段
        );

        //消息的同步确认 调用send方法发送消息
        final Future<RecordMetadata> future = producer.send(record);
        //调用get方法接收消息
        final RecordMetadata metadata = future.get();
        System.out.println("消息的主题:" + metadata.topic());
        System.out.println("消息的分区:" + metadata.partition());
        System.out.println("消息的偏移量:" + metadata.offset());

        //消息的异步确认
        // producer.send(record, new Callback() {
        //     @Override
        //     public void onCompletion(RecordMetadata metadata, Exception exception) {
        //         if (exception == null) {
        //             System.out.println("消息的主题:" + metadata.topic());
        //             System.out.println("消息的分区:" + metadata.partition());
        //             System.out.println("消息的偏移量:" + metadata.offset());
        //         } else {
        //             System.out.println("异常消息");
        //         }
        //     }
        // });

        //关闭生产者
        producer.close();
    }
}
